package com.fudan.run.trans;

import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fudan.cfg.base.MapUtils;
import com.fudan.cfg.trans.ChangeFieldMappingTransDef;
import com.fudan.run.JobRunner;
import com.fudan.run.VariableRddMap;
import com.fudan.run.ctx.annotation.TransAdapter;
import com.lobstar.xrpc.service.utils.JsonUtils;

@TransAdapter("field_mapping")
public class ChangeFieldMappingTrans extends TransHandler<ChangeFieldMappingTransDef>{

	@Override
	public void trans(JobRunner jobRunner, ChangeFieldMappingTransDef transNode, VariableRddMap variableMap) {
		
		JavaRDD<Map<String,Object>> rdd = variableMap.getRdd(transNode.getTarget());
		JavaRDD<Map<String, Object>> resultRdd = rdd.map(new Function<Map<String,Object>, Map<String,Object>>() {
			@Override
			public Map<String, Object> call(Map<String, Object> record) throws Exception {
				List<Map<String,String>> mapping = transNode.getMapping();
				for (Map<String, String> map : mapping) {
					String fieldName = map.get("key");
					String formatter = map.get("format");
					Object value = MapUtils.getMapValue(record, fieldName);
					value = trans(value,formatter,map);
					MapUtils.putValue(record, fieldName, value,true);
				}
				return record;
			}
		});
		variableMap.addRdd(transNode.getName(), resultRdd);
	}
	
	public Object trans(Object value,String mapping,Map<String,?> def) {
		switch(mapping) {
		case "int":
			return (Integer)value;
		case "string":
			return (String)value;
		case "object":
			if(value == null || "".equals(value)) {
				return new HashMap<>();
			}
			if(Map.class.isAssignableFrom(value.getClass())) {
				return value;
			}
			try {
				value = JsonUtils.toObject(String.valueOf(value),new TypeReference<Map<String,Object>>() {});				
			}catch(Exception e) {
				value = "FORMAT_ERROR!";
			}
			return value;
		case "date":{
			try {
				String pf = MapUtils.getMapValue(def, "from");
				if(pf != null && !pf.trim().equals("")) {
					LocalDateTime ldt = LocalDateTime.parse(String.valueOf(value), DateTimeFormatter.ofPattern(pf));
					String af = MapUtils.getMapValue(def, "to");
					if(af == null || af.equals("")) {
						af = "yyyy-MM-dd HH:mm:ss";
					}
					String timeValue = ldt.format(DateTimeFormatter.ofPattern(af));
					return timeValue;
				}else {
					try {
						Date date = new Date(Long.parseLong((String)value));
						LocalDateTime ldt =LocalDateTime.ofInstant(date.toInstant(), ZoneId.systemDefault());
						String af = MapUtils.getMapValue(def, "to");
						if(af == null || af.equals("")) {
							af = "yyyy-MM-dd HH:mm:ss";
						}
						String timeValue = ldt.format(DateTimeFormatter.ofPattern(af));
						return timeValue;
					}catch(Exception e) {
						//skip
						return "NAN";
					}
				}
			}catch(Exception e) {
				return "NAN";
			}
		}
			default:
				return value;
		}
	}

}
